package com.leaf.spring.y201907.day30.kafka;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * 生产者
 * 
 * @Title:
 * @Description:
 * @Author: leaf
 * @Since:2019年7月30日
 * @Version:1.1.0
 */
public class TestProducter {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		Properties properties = new Properties();
		// 指定kafka服务器地址，如果是集群可以指定多个，但是就算只指定一个他也会去集群环境下寻找其他的节点地址
		properties.setProperty("bootstrap.servers", "192.168.89.40:9092");
		// key序列化器
		properties.setProperty("key.serializer", StringSerializer.class.getName());
		// value 序列化器
		properties.setProperty("value.serializer", StringSerializer.class.getName());

		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

		ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", 1, "testKey", "hello");
		Future<RecordMetadata> send = kafkaProducer.send(record);
		RecordMetadata recordMetadata = send.get();
		System.out.println("消息所在分区:" + recordMetadata.partition() + ",偏移量：" + recordMetadata.offset() + ",topic=" + recordMetadata.topic());
	}

}
